package org.mx.spring.utils;

import java.util.HashMap;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.TimeUnit;

import javax.validation.constraints.NotNull;

import org.mx.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.redis.core.RedisTemplate;

/**
 * 基于Redis实现的集群节点注册和发现工具类
 * 
 * @author john.peng
 *
 */
public class ClusterNodeDiscoverUtils {
	private static final Logger logger = LoggerFactory.getLogger(ClusterNodeDiscoverUtils.class);
	private RedisTemplate<String, String> redisTemplate;

	/**
	 * 构造函数
	 * @param redisTemplate RedisTemplate
	 */
	public ClusterNodeDiscoverUtils(RedisTemplate<String, String> redisTemplate) {
		super();
		this.redisTemplate = redisTemplate;
	}

	/**
	 * 根据指定的应用归属获取缓存的集群节点集合
	 * @param owner 应用归属，标识某一个特定应用
	 * @return 集群节点集合，key为节点ID，value为集群连接点
	 */
	public Map<String, String> getEndpoints(@NotNull String owner) {
		Map<String, String> endpoints = new HashMap<>();
		String clusterKey = String.format("cluster.keys.%s", owner);
		Set<String> ownerKeys = redisTemplate.boundSetOps(clusterKey).members();
		for (String ownerKey : ownerKeys) {
			String endpoint = redisTemplate.boundValueOps(ownerKey).get();
			if (StringUtils.isBlank(endpoint)) {
				// 指定的key已经过期
				redisTemplate.boundSetOps(clusterKey).remove(ownerKey);
				if (logger.isDebugEnabled()) {
					logger.debug(String.format("The node endpoint is expired, owner: %s, id: %s, it will be removed.",
							owner, ownerKey));
				}
			} else {
				// 取出真正的id，去掉redis中存储的前缀
				ownerKey = ownerKey.substring(clusterKey.length() + 1);
				endpoints.put(ownerKey, endpoint);
			}
		}
		return endpoints;
	}

	/**
	 * 注册一个节点
	 * @param owner 	应用归属，标识某一个特定应用
	 * @param id 		集群节点关联ID，标识某一个特定节点
	 * @param endpoint  集群连接点
	 */
	public void registryNode(@NotNull String owner, @NotNull String id, @NotNull String endpoint) {
		String clusterKey = String.format("cluster.keys.%s", owner);
		String ownerKey = String.format("%s.%s", clusterKey, id);
		redisTemplate.boundSetOps(clusterKey).add(ownerKey);
		redisTemplate.boundValueOps(ownerKey).set(endpoint, 1, TimeUnit.MINUTES);
		if (logger.isDebugEnabled()) {
			logger.debug(
					String.format("Registry a node success, owner: %s, id: %s, endpoint: %s.", owner, id, endpoint));
		}
	}
}
